package com.intct.hbase.func;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author gufg
 * @since 2025-08-12 18:32
 */
public class ProLoadDimMap extends RichMapFunction<Tuple2<String,Integer>,Tuple2<String,String>> {
    private static final Logger LOG = LoggerFactory.getLogger(ProLoadDimMap.class.getName());

    ScheduledExecutorService executor = null;
    private Map<String,String> cache;

    @Override
    public void open(Configuration parameters) throws Exception {
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    load();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },5,5, TimeUnit.MINUTES);//每隔 5 分钟拉取一次维表数据
    }

    @Override
    public void close() throws Exception {

    }
    @Override
    public Tuple2<String, String> map(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
        String username = stringIntegerTuple2.f0;
        Integer city_id = stringIntegerTuple2.f1;
        String cityName = cache.get(city_id.toString());
        return new Tuple2<String,String>(username,cityName);
    }

    public void load() throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        Connection con = DriverManager.getConnection("jdbc:mysql://172.XX.XX.XX:XX06/yumd?useSSL=false&characterEncoding=UTF-8", "root", "Root@123");
        PreparedStatement statement = con.prepareStatement("select city_id,city_name from city_dim;");
        ResultSet rs = statement.executeQuery();
        //全量更新维度数据到内存
        while (rs.next()) {
            String cityId = rs.getString("city_id");
            String cityName = rs.getString("city_name");
            cache.put(cityId, cityName);
        }
        con.close();
    }
}